package com.cube.reactive.guide.reactive;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

/**
 * @author litb
 * @since 2024/6/13 17:37
 */
public class SubscribeOnAndPublishOn {

    public static void main(String[] args) {
        Scheduler s = Schedulers.newParallel("subscribe-scheduler", 4);
        Scheduler d = Schedulers.newParallel("publish-scheduler", 4);

        //预期输出
        //Map1: 1 on thread subscribe-scheduler-1
        //Map1: 2 on thread subscribe-scheduler-1
        //Map1: 3 on thread subscribe-scheduler-1
        //Map1: 4 on thread subscribe-scheduler-1
        //Map1: 5 on thread subscribe-scheduler-1
        //Map2: 1 on thread publish-scheduler-2
        //Received: 1 on thread publish-scheduler-2
        //Map2: 2 on thread publish-scheduler-2
        //Received: 2 on thread publish-scheduler-2
        //Map2: 3 on thread publish-scheduler-2
        //Received: 3 on thread publish-scheduler-2
        //Map2: 4 on thread publish-scheduler-2
        //Received: 4 on thread publish-scheduler-2
        //Map2: 5 on thread publish-scheduler-2
        //Received: 5 on thread publish-scheduler-2

        //subscribeOn(s) 确保源（Flux.range(1, 5)) 和第一个 map 操作在 subscribe-scheduler 的线程上执行。
        //publishOn(d) 切换线程，使得后续的 map 操作和 subscribe 操作在 publish-scheduler 的线程上执行。
        Flux<Integer> flux = Flux.range(1, 5)
                .map(i -> {
                    System.out.println("Map1: " + i + " on thread " + Thread.currentThread().getName());
                    return i;
                })
                .publishOn(d)
                .map(i -> {
                    System.out.println("Map2: " + i + " on thread " + Thread.currentThread().getName());
                    return i;
                })
                .subscribeOn(s);

        flux.subscribe(i -> System.out.println("Received: " + i + " on thread " + Thread.currentThread().getName()));

    }
}
